from kafka import KafkaProducer


class ProducerClass:
    def __init__(
            self,
            ip_address: str,
            topic: str,
            key: str,
            msg: any,
            headers: str
    ):
        self.ip_address = ip_address
        self.topic = topic
        self.key = key
        self.msg = msg
        self.headers = headers

    def send_msg_producer(self):
        kafka_producer = KafkaProducer(bootsrtap_servers=[self.ip_address], api_version=(0, 10))

        kafka_producer.send(topic=self.topic, key=self.key.encode(), value=self.msg, headers=self.headers)
        kafka_producer.flush()
        if kafka_producer is not None:
            kafka_producer.close()
